package HandshakeJoin;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import com.google.common.base.Preconditions;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.Stream;

public class JoinDriver extends ProcessingElement
{
    private Stream<Message> downstream;
    private Message.Direction direction;
    private long tupleLifespan = 1000;
    private ArrayList<JoinConfig> joinConfigs = null;
    private long seq;

    @Override
    protected void onCreate() {
        seq = 0;
    }

    public void onEvent(DataEvent event) {
        Tuple tuple0 = new Tuple(event.getTuple(), generateSeq());

        for (JoinConfig jConfig : joinConfigs) {
            Tuple tuple = new Tuple(tuple0);
            if (direction == Message.Direction.FROM_LEFT) {
                tuple.setTargetFieldIdx(jConfig.getJoinFieldInR());
            }
            else if (direction == Message.Direction.FROM_RIGHT) {
                tuple.setTargetFieldIdx(jConfig.getJoinFieldInS());
            }
            else {
                // dummy
            }
            tuple.setLifespan(tupleLifespan);

            Message msg = new Message(jConfig.getId(), direction, tuple);

            downstream.put(msg);
        }
    }

    public void onTime() {
        for (JoinConfig jConfig : joinConfigs) {
            Message msg = new Message(jConfig.getId(), Message.Type.ACK,
                    direction, 0);

            downstream.put(msg);
        }
    }

    @Override
    protected void onRemove() {}

    private long generateSeq() {
        return seq++;
    }

    public JoinDriver setDownStream(Stream<Message> stream) {
        this.downstream = Preconditions.checkNotNull(stream);

        return this;
    }

    public JoinDriver setDirection(final Message.Direction direction) {
        this.direction = direction;

        return this;
    }

    public JoinDriver setJoinConfigs(final ArrayList<JoinConfig> jConfigs) {
        this.joinConfigs = jConfigs;

        return this;
    }

    public JoinDriver setExpiryInterval(final long interval) {
        setTimerInterval(interval, TimeUnit.MILLISECONDS);

        return this;
    }

    public JoinDriver setTupleLifespan(final long lifespan) {
        this.tupleLifespan = lifespan;

        return this;
    }

    public JoinDriver setDriverName(final String name) {
        setName(name);

        return this;
    }
}